﻿using IOP.Net;
using IOP.Models.Queue;
using System;
using System.Buffers;
using System.Collections.Generic;
using System.Net.Sockets;
using System.Text;
using System.Threading.Tasks;

namespace IOP.Pulsar.Abstractions
{
    /// <summary>
    /// 默认管线入口
    /// </summary>
    public class DefaultProductLineEntry : IProductLineEntry<StringContext>
    {
        /// <summary>
        /// 包队列
        /// </summary>
        public LoopQueue<StringContext> Packages { get; set; } = new LoopQueue<StringContext>();

        /// <summary>
        /// 管线
        /// </summary>
        public ProductLineDelegate<StringContext> ProductLine { get; set; }

        /// <summary>
        /// 连接
        /// </summary>
        public SocketMonitor ConnectedSocket { get; set; }

        /// <summary>
        /// 构造函数
        /// </summary>
        public DefaultProductLineEntry(IProductLine<StringContext> productLine)
        {
            if (productLine == null) throw new ArgumentNullException(nameof(productLine));
            ProductLine = productLine.ProductLine;
            Packages.OnQueueHasElements += Packages_OnQueueHasElements;
        }
        /// <summary>
        /// 构造函数
        /// </summary>
        /// <param name="productLine"></param>
        public DefaultProductLineEntry(ProductLineDelegate<StringContext> productLine)
        {
            if (productLine == null) throw new ArgumentNullException(nameof(productLine));
            ProductLine = productLine;
            Packages.OnQueueHasElements += Packages_OnQueueHasElements;
        }


        /// <summary>
        /// 入口处理函数
        /// </summary>
        /// <param name="queue">队列</param>
        /// <param name="socket">当前嵌套字</param>
        /// <returns></returns>
        public void EntryHandle(ref ReadOnlySequence<byte> queue, SocketMonitor socket)
        {
            if (ConnectedSocket == null) ConnectedSocket = socket;
            SequencePosition? position;
            do
            {
                position = queue.PositionOf((byte)'\n');
                if (position != null)
                {
                    var line = queue.Slice(0, position.Value);
                    BuildContext(line, socket.ConnectedSocket);
                    // This is equivalent to position + 1
                    var next = queue.GetPosition(1, position.Value);
                    // Skip what we've already processed including
                    queue = queue.Slice(next);
                }
            } while (position != null);
        }

        /// <summary>
        /// 构建上下文
        /// </summary>
        /// <param name="buffer"></param>
        /// <param name="socket">当前嵌套字</param>
        private Task BuildContext(in ReadOnlySequence<byte> buffer, Socket socket)
        {
            foreach (var segment in buffer)
            {
                var result = Encoding.UTF8.GetString(segment.Span.ToArray());
                var context = new StringContext
                {
                    Body = result,
                    Socket = socket
                };
                Packages.Enqueue(context);
            }
            return Task.CompletedTask;
        }

        /// <summary>
        /// 队列事件
        /// </summary>
        /// <param name="obj"></param>
        private void Packages_OnQueueHasElements(LoopQueue<StringContext> obj)
        {
            while (!obj.IsEmpty)
            {
                obj.Dequene(out StringContext context);
                ProductLine?.Invoke(context);
            }
        }

        /// <summary>
        /// 释放资源
        /// </summary>
        public void Dispose()
        {
            Packages.OnQueueHasElements -= Packages_OnQueueHasElements;
            ConnectedSocket.ConnectedSocket.Disconnect(false);
            ConnectedSocket.ConnectedSocket.Shutdown(SocketShutdown.Both);
            ConnectedSocket.ConnectedSocket.Close();
            ConnectedSocket.ConnectedSocket.Dispose();
        }
    }
}
